package wangwenjun.phase2.concurrencypattern.eventbus.framework;

import lombok.extern.slf4j.Slf4j;
import util.ThreadUtil;

import java.lang.reflect.Method;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;

/**
 * @author ChangLiang
 * @date 2020/6/18
 */
@Slf4j
class EventDispatcher {

    /**
     * Dispatcher 的主要作用是将EventBus post 的event 推送给每一个注册到topic 上的subscriber 上，
     * 具体的推送其实就是执行被@EventSubscribe 注解的方法
     */

    private final Executor executorService;

    private final EventExceptionHandler exceptionHandler;

    public static final Executor SEQ_EXECUTOR_SERVICE = SeqExecutorService.INSTANCE;

    public static final Executor PER_THREAD_EXECUTOR_SERVICE = PerThreadExecutorService.INSTANCE;

    private EventDispatcher(Executor executorService, EventExceptionHandler exceptionHandler) {
        this.executorService = executorService;
        this.exceptionHandler = exceptionHandler;
    }

    public void dispatchEvent(EventBus eventBus, SubscriberRegistry registry, Object event, String topic) {
        ConcurrentLinkedQueue<Subscriber> subscribers = registry.getSubscribersByTopic(topic);
        if (null == subscribers) {
            if (exceptionHandler != null) {
                exceptionHandler.handle(
                        new IllegalArgumentException("the topic " + topic + " has not bind yet"),
                        new BaseEventContext(eventBus.getEventBusName(), null, event));
            }
            return;
        }
        subscribers.stream()
                .filter(subscriber -> !subscriber.isDisable())
                .filter(subscriber -> {
                    Method subscribeMethod = subscriber.getSubscribeMethod();
                    // 获取到第一个参数的所属类
                    Class<?> aClass = subscribeMethod.getParameterTypes()[0];
                    // 所属类是否是event类，或其子类
                    return aClass.isAssignableFrom(event.getClass());
                    // foreach 就是使用线程池的worker thread，通过反射来执行回调方法
                }).forEach(subscriber -> invokeSubscribeMethod(subscriber,event,eventBus));
    }

    private void invokeSubscribeMethod(Subscriber subscriber, Object event, EventBus eventBus) {
        Object subscribeObject = subscriber.getSubscribeObject();
        Method subscribeMethod = subscriber.getSubscribeMethod();
        /**
         * 执行回调方法
         */
        executorService.execute(()->{
            try {
                subscribeMethod.invoke(subscribeObject, event);
            } catch (Exception e) {
                // 是在其他线程中执行过程中报错 这里是可以catch到的
                // [pool-1-thread-4] [ERROR] -- java.lang.reflect.InvocationTargetException
                log.error(ThreadUtil.exceptionToString(e));
                this.exceptionHandler.handle(e, new BaseEventContext(eventBus.getEventBusName(), subscriber, event));
            }
        });
    }

    public void close() {
        if (executorService instanceof ExecutorService) {
            ((ExecutorService) executorService).shutdown();
        }
    }

    static EventDispatcher newEventDispatcher(EventExceptionHandler exceptionHandler, Executor executor) {
        return new EventDispatcher(executor, exceptionHandler);
    }

    private static class SeqExecutorService implements Executor {

        private final static SeqExecutorService INSTANCE = new SeqExecutorService();

        @Override
        public void execute(Runnable command) {
            command.run();
        }
    }

    private static class PerThreadExecutorService implements Executor {

        private final static PerThreadExecutorService INSTANCE = new PerThreadExecutorService();

        @Override
        public void execute(Runnable command) {
            new Thread(command).start();
        }
    }

    private static class BaseEventContext implements EventContext {

        private final String eventBusName;

        private final Subscriber subscriber;

        private final Object event;

        private BaseEventContext(String eventBusName, Subscriber subscriber, Object event) {
            this.eventBusName = eventBusName;
            this.subscriber = subscriber;
            this.event = event;
        }


        @Override
        public String getEventBusName() {
            return this.eventBusName;
        }

        @Override
        public Object getSubscriber() {
            return subscriber != null ? subscriber.getSubscribeObject() : null;
        }

        @Override
        public Method getMethod() {
            return subscriber != null ? subscriber.getSubscribeMethod() : null;
        }

        @Override
        public Object getEvent() {
            return this.event;
        }
    }

}
